from pyflink.common import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, MapFunction
from pyflink.datastream.connectors import DeliveryGuarantee
from pyflink.datastream.connectors.kafka import KafkaSink, KafkaRecordSerializationSchema

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 修改并行度
env.set_parallelism(1)

# 2、读取数据,得到DataStream，相当于RDD  （有界流）
lines_ds = env.read_text_file("../../data/students.txt")

# 构建kafka sink
sink = KafkaSink.builder() \
    .set_bootstrap_servers("master:9092") \
    .set_record_serializer(
    KafkaRecordSerializationSchema.builder()
    .set_topic("students_csv")
    .set_value_serialization_schema(SimpleStringSchema())
    .build()) \
    .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE) \
    .build()

lines_ds.sink_to(sink)

env.execute()
